package com.atguigu.champter5.Flink05.test;

import com.atguigu.champter5.Flink05.test.bean.AdsClickLog;
import com.atguigu.champter5.Flink05.test.bean.OrderEvent;
import com.atguigu.champter5.Flink05.test.bean.TxEvent;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.co.CoMapFunction;
import org.apache.flink.streaming.api.functions.co.CoProcessFunction;
import org.apache.flink.util.Collector;
import org.apache.hadoop.yarn.webapp.hamlet2.Hamlet;

import java.util.HashMap;

public class Flink_Match {
    public static void main(String[] args) {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        DataStreamSource<String> orderdss = env.readTextFile("datas/input/OrderLog.csv");
        DataStreamSource<String> recedss = env.readTextFile("datas/input/ReceiptLog.csv");


        SingleOutputStreamOperator<OrderEvent> orderMap = orderdss.map(new MapFunction<String, OrderEvent>() {
            @Override
            public OrderEvent map(String value) throws Exception {
                String[] data = value.split(",");
                return new OrderEvent(
                        Long.valueOf(data[0]),
                        String.valueOf(data[1]),
                        String.valueOf(data[2]),
                        Long.valueOf(data[3])
                );
            }
        });

        SingleOutputStreamOperator<TxEvent> txMap = recedss.map(new MapFunction<String, TxEvent>() {
            @Override
            public TxEvent map(String value) throws Exception {
                String[] data = value.split(",");
                return new TxEvent(
                        String.valueOf(data[0]),
                        String.valueOf(data[1]),
                        Long.valueOf(data[2])
                );
            }
        });

        orderMap.connect(txMap)
                .keyBy("txId","txId")
                .process(new CoProcessFunction<OrderEvent, TxEvent, String>() {
                    HashMap<String,TxEvent> txmap= new HashMap<>();
                    HashMap<String,OrderEvent> ormap= new HashMap<>();
                    @Override
                    public void processElement1(OrderEvent value, CoProcessFunction<OrderEvent, TxEvent, String>.Context ctx, Collector<String> out) throws Exception {
                        if (txmap.containsKey(value.getTxId())){
                            out.collect("订单:"+value.getOrderId() + "支付成功");
                            txmap.remove(value.getTxId());
                        }else{
                            ormap.put(value.getTxId(),value);
                        }

                    }

                    @Override
                    public void processElement2(TxEvent value, CoProcessFunction<OrderEvent, TxEvent, String>.Context ctx, Collector<String> out) throws Exception {
                        if (ormap.containsKey(value.getTxId())){
                            out.collect("订单:"+ormap.get(value.getTxId()).getOrderId() + "支付成功");
                            ormap.remove(value.getTxId());
                        }else {
                            txmap.put(value.getTxId(),value);
                        }

                    }
                }).print();

        try {
            env.execute();
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}
